package com.ruoyi.im.listener;

import com.alibaba.nacos.common.utils.CollectionUtils;
import com.ruoyi.common.core.domain.R;
import com.ruoyi.im.dao.MessageDAO;
import com.ruoyi.im.domain.Message;
import com.ruoyi.im.utils.JsonUtils;
import com.ruoyi.im.ws.MessageHandler;
import com.ruoyi.system.api.RemoteUserService;
import com.ruoyi.system.api.domain.SysUser;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.List;
import java.util.Map;

/**
 * 〈公告监听器〉
 *
 * @author coach tam
 * @email 327395128@qq.com
 * @create 2023/10/25
 * @since 1.0.0
 * 〈坚持灵活 灵活坚持〉
 */
@Component
@Slf4j
public class NoticeListener {

    @Autowired
    private MessageHandler messageHandler;
    @Autowired
    private MessageDAO messageDAO;

    @Autowired
    private RemoteUserService userService;

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "system.notice.queue"),
            exchange = @Exchange(name = "system.notice.exchange", type = ExchangeTypes.TOPIC),
            key = "notice.insert"
    ))
    public void insert(Map<String, Object> data) {
        if (MapUtils.isEmpty(data) && data.size() != 3) {
            return;
        }
        String msg = data.get("msg").toString();
        String from = data.get("from").toString();
        Long businessId = Long.valueOf(data.get("businessId").toString());
        //给所有用户保存未读消息
        R<List<SysUser>> result = userService.getAll();
        int code = result.getCode();
        if (code != HttpStatus.OK.value()) {
            log.error(result.getMsg());
            return;
        }
        List<SysUser> users = result.getData();
        if (CollectionUtils.isEmpty(users)) {
            return;
        }
        Message pojo = Message.builder()
                .from(from)
                .status(1)
                .msg(msg)
                .businessId(businessId)
                .build();
        users.forEach(user->
                {
                    pojo.setTo(user.getUserName());
                    messageDAO.saveMessage(pojo);
                }
        );

        //获取给所有在线用户推送消息
        Map<String, WebSocketSession> sessions = messageHandler.getSessions();
        if (MapUtils.isEmpty(sessions)) {
            return;
        }
        //构建一个公共的推送消息对象
        Message message = Message.builder()
                .from(from)
                .msg(msg)
                .build();
        for (Map.Entry<String, WebSocketSession> entry : sessions.entrySet()) {
            WebSocketSession session = entry.getValue();
            try {
                String toId = entry.getKey();
                messageDAO.updateMessageState(businessId, toId, 2);
                session.sendMessage(new
                        TextMessage(JsonUtils.serialize(message)));
            } catch (IOException e) {
                log.error("{}给{}发送即时消息失败.", message.getFrom(), message.getTo());
            }
        }
    }
}
